package com.zhengbo.flink.demo.base.codis;

import com.caocao.bridge.codis.proxy.jodis.CodisResourcePool;
import com.caocao.bridge.codis.proxy.jodis.RoundRobinCodisPool;
import lombok.Data;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.util.Pool;

@Component("codisClient")
@Data
public class CodisClient {

    private static Logger logger = LoggerFactory.getLogger(CodisClient.class);

    private volatile CodisResourcePool codisResourcePool;

    @Value("${codis.proxy.codisName}")
    private String codisName;

    @Value("${codis.proxy.password}")
    private String codisPassword;

    @Value("${codis.proxy.zkServer}")
    private String codisZKServer;

    private CodisResourcePool getCodisResourcePool(){
        if (codisResourcePool == null) {
            synchronized (this) {
                if (codisResourcePool == null) {
                    codisResourcePool = RoundRobinCodisPool.create()
                            .curatorClient(codisZKServer, 30000)
                            .zkProxyDir("/jodis/" + codisName)
                            .password(codisPassword)
                            .poolConfig(new JedisPoolConfig())
                            .build();
                }
            }
        }
        return codisResourcePool;
    }

    public Pool<Jedis> getJedisPool(){
        return getCodisResourcePool().getPool();
    }

    public void returnResource(Pool<Jedis> jedisPool, Jedis jedis){
        if ( jedisPool != null && jedis != null) {
            getCodisResourcePool().returnResource(jedisPool, jedis);
        }
    }
}